//
// Created by v on 19-6-27.
//

#include "CMMqttMosquitto.h"
#include <errno.h>
#include <stdio.h>
#include <string.h>

#ifdef USE_MQTT_MOSQ

void cm_mqtt_init_all() { mosquitto_lib_init(); }

void cm_mqtt_deinit_all() { mosquitto_lib_cleanup(); }

void log_callback(struct mosquitto *mosq, void *obj, int level, const char *str)
{
    CMMqttFunction *f = (CMMqttFunction *)obj;
    f->onLog(level, str);
}

void publish_callback(struct mosquitto *mosq, void *obj, int mid)
{
    CMMqttFunction *f = (CMMqttFunction *)obj;
    f->onPublish(mid, 0);
}

void message_callback(struct mosquitto *mosq, void *obj,
                      const struct mosquitto_message *message)
{
    CMMqttFunction *f = (CMMqttFunction *)obj;
    f->onMessage(message->mid, message->topic, (const char *)message->payload,
                 message->payloadlen);
}

void connect_callback(struct mosquitto *mosq, void *obj, int result)
{
    CMMqttFunction *f = (CMMqttFunction *)obj;
    f->onConnect(result == MOSQ_ERR_SUCCESS ? 0 : -result);
}

void disconnect_callback(struct mosquitto *mosq, void *obj, int rc)
{
    CMMqttFunction *f = (CMMqttFunction *)obj;
    f->onDisconnect(rc);
}

void subscribe_callback(struct mosquitto *mosq, void *obj, int mid,
                        int qos_count, const int *granted_qos)
{
    CMMqttFunction *f = (CMMqttFunction *)obj;
    f->onSubscribe(mid);
}

#ifdef MOSQ_NEW
void publish_v5_callback(struct mosquitto *mosq, void *obj, int mid,
                         int reason_code, const mosquitto_property *properties)
{
    CMMqttFunction *f = (CMMqttFunction *)obj;
    f->onPublish(mid, reason_code);
}

void message_v5_callback(struct mosquitto *mosq, void *obj,
                         const struct mosquitto_message *message,
                         const mosquitto_property *      properties)
{
    CMMqttFunction *f = (CMMqttFunction *)obj;
    f->onMessage(message->mid, message->topic, (const char *)message->payload,
                 message->payloadlen);
}

void connect_v5_callback(struct mosquitto *mosq, void *obj, int result,
                         int flags, const mosquitto_property *properties)
{
    CMMqttFunction *f = (CMMqttFunction *)obj;
    f->onConnect(result);
}

void disconnect_v5_callback(struct mosquitto *mosq, void *obj, int rc,
                            const mosquitto_property *properties)
{
    CMMqttFunction *f = (CMMqttFunction *)obj;
    f->onDisconnect(rc);
}

void subscribe_v5_callback(struct mosquitto *mosq, void *obj, int mid,
                           int qos_count, const int *granted_qos,
                           const mosquitto_property *)
{
    CMMqttFunction *f = (CMMqttFunction *)obj;
    f->onSubscribe(mid);
}

#endif

CMMqttMosquitto::CMMqttMosquitto(const char *host, int port)
    : CMMqttBase(host, port)
{
    m_mosq = NULL;
#ifdef MOSQ_NEW
    m_willProps = NULL;
#endif
    m_keepAlive    = 300;
    m_sessionClean = true;
    m_mqtt_v5      = false;
    m_bindAddress  = "0.0.0.0";
    m_connected    = false;
    m_cafile.clear();
    m_capath.clear();
    m_certfile.clear();
    m_keyfile.clear();
}

CMMqttMosquitto::~CMMqttMosquitto() { deinit(); }

int CMMqttMosquitto::init()
{
    if (!m_mosq)
    {
        m_mosq = mosquitto_new(m_id.c_str(), m_sessionClean, m_function);
        if (!m_mosq)
        {
            return errno;
        }
    }
    if (m_debug)
    {
        mosquitto_log_callback_set(m_mosq, log_callback);
    }
    if (m_mqtt_v5)
    {
#ifdef MOSQ_NEW
        mosquitto_connect_v5_callback_set(m_mosq, connect_v5_callback);
        mosquitto_disconnect_v5_callback_set(m_mosq, disconnect_v5_callback);
        mosquitto_publish_v5_callback_set(m_mosq, publish_v5_callback);
        mosquitto_message_v5_callback_set(m_mosq, message_v5_callback);
        mosquitto_subscribe_v5_callback_set(m_mosq, subscribe_v5_callback);
#endif
    }
    else
    {
        mosquitto_connect_callback_set(m_mosq, connect_callback);
        mosquitto_disconnect_callback_set(m_mosq, disconnect_callback);
        mosquitto_publish_callback_set(m_mosq, publish_callback);
        mosquitto_message_callback_set(m_mosq, message_callback);
        mosquitto_subscribe_callback_set(m_mosq, subscribe_callback);
    }
    static int v = MQTT_PROTOCOL_V311;
    mosquitto_opts_set(m_mosq, MOSQ_OPT_PROTOCOL_VERSION, (void *)&v);

    mosquitto_max_inflight_messages_set(m_mosq, 20);

    if (m_withTls)
    {
        int ret = mosquitto_tls_set(
            m_mosq, m_cafile.empty() ? NULL : m_cafile.c_str(),
            m_capath.empty() ? NULL : m_capath.c_str(),
            m_certfile.empty() ? NULL : m_certfile.c_str(),
            m_keyfile.empty() ? NULL : m_keyfile.c_str(), NULL);
        if (m_debug)
        {
            char str[1024] = {0};
            sprintf(str, "with tls ret = [%d]", ret);
            if (m_function)
                m_function->onLog(4, str);
        }
    }
    if (m_insecureTls)
    {
        int ret = mosquitto_tls_insecure_set(m_mosq, true);
        if (m_debug)
        {
            char str[1024] = {0};
            sprintf(str, "with tls insecure = [%d]", ret);
            if (m_function)
                m_function->onLog(4, str);
        }
    }
    if (m_debug)
    {
        char str[1024] = {0};
        sprintf(str, "m_mosq = [%p]", m_mosq);
        if (m_function)
            m_function->onLog(4, str);
    }
    return 0;
}

int CMMqttMosquitto::deinit()
{
    if (m_mosq)
    {
        mosquitto_destroy(m_mosq);
    }
    m_mosq = NULL;
    return 0;
}

int CMMqttMosquitto::connect()
{
    //    int ret = mosquitto_connect_srv(m_mosq, m_host.c_str(), m_keepAlive,
    //    m_bindAddress.c_str());
    if (m_debug)
    {
        char str[1024] = {0};
        snprintf(str, sizeof(str), "connect to url [%s:%d]\n", m_host.c_str(),
                 m_port);
        if (m_function)
            m_function->onLog(4, str);
    }
    int ret = mosquitto_connect_bind_async(m_mosq, m_host.c_str(), m_port,
                                           m_keepAlive, NULL);
    if (ret == MOSQ_ERR_SUCCESS)
    {
        ret = mosquitto_loop_start(m_mosq);
    }
    else
    {
        if (m_debug)
        {
            char errstr[1024] = {0};
            snprintf(errstr, sizeof(errstr), "connect fail [%s]\n",
                     strerror(errno));
            if (m_function)
                m_function->onLog(4, errstr);
        }
    }
    return ret > MOSQ_ERR_SUCCESS ? -ret : ret;
}

int CMMqttMosquitto::reconnect()
{
    int ret = mosquitto_reconnect_async(m_mosq);
    return ret > MOSQ_ERR_SUCCESS ? -ret : ret;
}

int CMMqttMosquitto::disconnect()
{
    mosquitto_loop_stop(m_mosq, true);
    int ret = mosquitto_disconnect(m_mosq);
    return ret > MOSQ_ERR_SUCCESS ? -ret : ret;
}

int CMMqttMosquitto::loop()
{
    int ret = mosquitto_loop(m_mosq, -1, 1);
    return ret > MOSQ_ERR_SUCCESS ? -ret : ret;
}

int CMMqttMosquitto::loop_block()
{
    int ret = mosquitto_loop_forever(m_mosq, -1, 1);
    return ret > MOSQ_ERR_SUCCESS ? -ret : ret;
}

int CMMqttMosquitto::setUsername(const char *user, const char *pass)
{
    m_username = user;
    m_password = pass;
    if (m_debug)
    {
        char errstr[1024] = {0};
        snprintf(errstr, sizeof(errstr), "set username handle [%p] [%s, %s]\n",
                 m_mosq, user, pass);
        if (m_function)
            m_function->onLog(4, errstr);
    }
    int ret    = mosquitto_username_pw_set(m_mosq, user, pass);
    return ret > MOSQ_ERR_SUCCESS ? -ret : ret;
}

int CMMqttMosquitto::setWill(const char *topic, const char *data)
{
    m_willTopic  = topic;
    m_willData   = data;
    m_willQoS    = 1;
    m_willRetain = true;
    int ret      = 0;
    if (m_mqtt_v5)
    {
#ifdef MOSQ_NEW
        ret = mosquitto_will_set_v5(m_mosq, topic, m_willData.length(), data,
                                    m_willQoS, m_willRetain, m_willProps);
#endif
    }
    else
    {
        ret = mosquitto_will_set(m_mosq, topic, m_willData.length(), data,
                                 m_willQoS, m_willRetain);
    }
    return ret > MOSQ_ERR_SUCCESS ? -ret : ret;
}

int CMMqttMosquitto::sub(const char *topic, int qos)
{
    int mid = 5;
    int ret = mosquitto_subscribe(m_mosq, &mid, topic, qos);
    return ret > MOSQ_ERR_SUCCESS ? -ret : ret;
}

int CMMqttMosquitto::pub(const char *topic, const char *data, int qos)
{
    int ret = mosquitto_publish(m_mosq, NULL, topic, strlen(data), data,
                                qos, m_willRetain);
    return ret > MOSQ_ERR_SUCCESS ? -ret : ret;
}

#else

CMMqttMosquitto::CMMqttMosquitto(const char *host, int port)
    : CMMqttBase(host, port)
{
}

void cm_mqtt_init_all() {}

void cm_mqtt_deinit_all() {}
#endif
